// DriverBehaviorMapper.java
package com.taxi.analysis.driver;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 司机行为分析Mapper
 * 功能：综合运用多种MapReduce技术分析司机行为
 * 技术：综合技术（分组+条件+排序+去重+分区+Join）
 */
public class DriverBehaviorMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();
    private Set<String> processedDriverTrips = new HashSet<>(); // 去重技术

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        if (key.get() == 0) return;

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) return;

            String[] fields = parseCSVLine(line);

            if (fields.length >= 26) {
                String driverId = fields[0].trim();
                String pickupDatetime = fields[1].trim();
                String pickupLocationId = fields[2].trim();
                String dropoffLocationId = fields[3].trim();
                int pickupHour = parseInt(fields[4]);
                int pickupWeekday = parseInt(fields[5]);
                String pickupDate = fields[6].trim();
                int pickupMonth = parseInt(fields[7]);
                double tripDistance = parseDouble(fields[8]);
                double totalAmount = parseDouble(fields[9]);
                double tripDuration = parseDouble(fields[10]);
                double avgSpeed = parseDouble(fields[11]);
                int passengerCount = parseInt(fields[12]);
                String workShift = fields[13].trim();
                String workDayType = fields[14].trim();
                String drivingStyle = fields[15].trim();
                String incomeLevel = fields[16].trim();
                String efficiencyLevel = fields[17].trim();
                double hourlyIncome = parseDouble(fields[18]);
                double incomePerMile = parseDouble(fields[19]);
                int routeFamiliarity = parseInt(fields[20]);
                String professionalismLevel = fields[21].trim();
                String workIntensity = fields[22].trim();
                int speedAnomaly = parseInt(fields[23]);
                int fareAnomaly = parseInt(fields[24]);
                int durationAnomaly = parseInt(fields[25]);
                int anomalyScore = parseInt(fields[26]);
                String anomalyLevel = fields[27].trim();

                if (isValidData(driverId, pickupLocationId, dropoffLocationId)) {

                    // 技术1: 分组技术 - 按司机分组分析
                    processDriverGrouping(driverId, pickupDatetime, pickupLocationId, dropoffLocationId,
                            pickupHour, workShift, workDayType, tripDistance, totalAmount,
                            tripDuration, avgSpeed, drivingStyle, incomeLevel, efficiencyLevel,
                            hourlyIncome, incomePerMile, routeFamiliarity, professionalismLevel,
                            workIntensity, anomalyLevel, context);

                    // 技术2: 条件过滤技术 - 筛选异常行为
                    processAnomalyFiltering(driverId, pickupDatetime, anomalyScore, anomalyLevel,
                            speedAnomaly, fareAnomaly, durationAnomaly, avgSpeed,
                            totalAmount, tripDuration, drivingStyle, context);

                    // 技术3: 排序技术 - 司机效率排名
                    processEfficiencyRanking(driverId, hourlyIncome, incomePerMile, efficiencyLevel,
                            professionalismLevel, workIntensity, tripDistance,
                            totalAmount, routeFamiliarity, context);

                    // 技术4: 去重技术 - 去除重复行程模式
                    processPatternDeduplication(driverId, pickupLocationId, dropoffLocationId,
                            pickupHour, workShift, drivingStyle, tripDistance,
                            totalAmount, context);

                    // 技术5: 分区技术 - 按工作模式分区
                    processWorkPatternPartitioning(driverId, workShift, workDayType, workIntensity,
                            pickupHour, pickupWeekday, drivingStyle,
                            efficiencyLevel, professionalismLevel, context);

                    // 技术6: Join准备 - 司机与地点关联
                    processDriverLocationJoin(driverId, pickupLocationId, dropoffLocationId,
                            routeFamiliarity, tripDistance, totalAmount,
                            hourlyIncome, professionalismLevel, context);
                }
            }

        } catch (Exception e) {
            System.err.println("处理行时出错: " + value.toString());
        }
    }

    /**
     * 技术1: 分组技术 - 按司机分组分析
     */
    private void processDriverGrouping(String driverId, String pickupDatetime, String pickupLocationId,
                                       String dropoffLocationId, int pickupHour, String workShift,
                                       String workDayType, double tripDistance, double totalAmount,
                                       double tripDuration, double avgSpeed, String drivingStyle,
                                       String incomeLevel, String efficiencyLevel, double hourlyIncome,
                                       double incomePerMile, int routeFamiliarity, String professionalismLevel,
                                       String workIntensity, String anomalyLevel, Context context)
            throws IOException, InterruptedException {

        String groupKey = "DRIVER_GROUP_" + driverId;
        String driverRecord = createDriverRecord(pickupDatetime, pickupLocationId, dropoffLocationId,
                pickupHour, workShift, workDayType, tripDistance,
                totalAmount, tripDuration, avgSpeed, drivingStyle,
                incomeLevel, efficiencyLevel, hourlyIncome, incomePerMile,
                routeFamiliarity, professionalismLevel, workIntensity, anomalyLevel);

        outputKey.set(groupKey);
        outputValue.set("DRIVER_BEHAVIOR|" + driverRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 技术2: 条件过滤技术 - 筛选异常行为
     */
    private void processAnomalyFiltering(String driverId, String pickupDatetime, int anomalyScore,
                                         String anomalyLevel, int speedAnomaly, int fareAnomaly,
                                         int durationAnomaly, double avgSpeed, double totalAmount,
                                         double tripDuration, String drivingStyle, Context context)
            throws IOException, InterruptedException {

        // 条件过滤：只处理异常行为
        if (anomalyScore >= 2 || "高异常".equals(anomalyLevel) || "中等异常".equals(anomalyLevel)) {
            String filterKey = "ANOMALY_FILTER_" + anomalyLevel;
            String anomalyRecord = createAnomalyRecord(driverId, pickupDatetime, anomalyScore,
                    speedAnomaly, fareAnomaly, durationAnomaly,
                    avgSpeed, totalAmount, tripDuration, drivingStyle);

            outputKey.set(filterKey);
            outputValue.set("ANOMALY_BEHAVIOR|" + anomalyRecord);
            context.write(outputKey, outputValue);
        }
    }

    /**
     * 技术3: 排序技术 - 司机效率排名
     */
    private void processEfficiencyRanking(String driverId, double hourlyIncome, double incomePerMile,
                                          String efficiencyLevel, String professionalismLevel,
                                          String workIntensity, double tripDistance, double totalAmount,
                                          int routeFamiliarity, Context context)
            throws IOException, InterruptedException {

        // 计算综合效率分数
        double efficiencyScore = calculateEfficiencyScore(hourlyIncome, incomePerMile, efficiencyLevel,
                professionalismLevel, workIntensity, routeFamiliarity);

        // 使用负数进行降序排序
        String rankKey = String.format("EFFICIENCY_RANK_%010.2f_%s", -efficiencyScore, driverId);
        String rankRecord = createRankRecord(driverId, hourlyIncome, incomePerMile, efficiencyLevel,
                professionalismLevel, workIntensity, tripDistance,
                totalAmount, routeFamiliarity, efficiencyScore);

        outputKey.set(rankKey);
        outputValue.set("EFFICIENCY_RANK|" + rankRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 技术4: 去重技术 - 去除重复行程模式
     */
    private void processPatternDeduplication(String driverId, String pickupLocationId, String dropoffLocationId,
                                             int pickupHour, String workShift, String drivingStyle,
                                             double tripDistance, double totalAmount, Context context)
            throws IOException, InterruptedException {

        // 生成行程模式唯一标识
        String patternKey = String.format("%s_%s_%s_%d_%s_%s", driverId, pickupLocationId,
                dropoffLocationId, pickupHour/2, workShift, drivingStyle);

        // 去重：只处理第一次出现的模式
        if (!processedDriverTrips.contains(patternKey)) {
            processedDriverTrips.add(patternKey);

            String dedupKey = "PATTERN_DEDUP_" + driverId;
            String patternRecord = createPatternRecord(pickupLocationId, dropoffLocationId,
                    pickupHour, workShift, drivingStyle,
                    tripDistance, totalAmount);

            outputKey.set(dedupKey);
            outputValue.set("UNIQUE_PATTERN|" + patternRecord);
            context.write(outputKey, outputValue);
        }
    }

    /**
     * 技术5: 分区技术 - 按工作模式分区
     */
    private void processWorkPatternPartitioning(String driverId, String workShift, String workDayType,
                                                String workIntensity, int pickupHour, int pickupWeekday,
                                                String drivingStyle, String efficiencyLevel,
                                                String professionalismLevel, Context context)
            throws IOException, InterruptedException {

        // 生成分区键
        String partitionKey = String.format("WORK_PARTITION_%s_%s_%s", workShift, workDayType, workIntensity);
        String workRecord = createWorkPatternRecord(driverId, pickupHour, pickupWeekday,
                drivingStyle, efficiencyLevel, professionalismLevel);

        outputKey.set(partitionKey);
        outputValue.set("WORK_PATTERN|" + workRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 技术6: Join准备 - 司机与地点关联
     */
    private void processDriverLocationJoin(String driverId, String pickupLocationId, String dropoffLocationId,
                                           int routeFamiliarity, double tripDistance, double totalAmount,
                                           double hourlyIncome, String professionalismLevel, Context context)
            throws IOException, InterruptedException {

        // 司机数据（用于Join）
        String driverJoinKey = "JOIN_DRIVER_" + driverId;
        String driverJoinRecord = createDriverJoinRecord(pickupLocationId, dropoffLocationId,
                routeFamiliarity, tripDistance, totalAmount,
                hourlyIncome, professionalismLevel);

        outputKey.set(driverJoinKey);
        outputValue.set("DRIVER_DATA|" + driverJoinRecord);
        context.write(outputKey, outputValue);

        // 地点数据（用于Join）
        String locationJoinKey = "JOIN_LOCATION_" + pickupLocationId;
        String locationJoinRecord = createLocationJoinRecord(driverId, dropoffLocationId,
                routeFamiliarity, tripDistance,
                totalAmount, professionalismLevel);

        outputKey.set(locationJoinKey);
        outputValue.set("LOCATION_DATA|" + locationJoinRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 计算综合效率分数
     */
    private double calculateEfficiencyScore(double hourlyIncome, double incomePerMile, String efficiencyLevel,
                                            String professionalismLevel, String workIntensity, int routeFamiliarity) {
        double score = 0.0;

        // 收入指标权重40%
        score += Math.min(hourlyIncome / 100.0, 1.0) * 20;
        score += Math.min(incomePerMile / 5.0, 1.0) * 20;

        // 效率等级权重25%
        switch (efficiencyLevel) {
            case "高效率": score += 25; break;
            case "中等效率": score += 15; break;
            default: score += 5; break;
        }

        // 专业度权重20%
        switch (professionalismLevel) {
            case "专业司机": score += 20; break;
            case "熟练司机": score += 12; break;
            default: score += 4; break;
        }

        // 工作强度权重10%
        switch (workIntensity) {
            case "高强度": score += 10; break;
            case "中等强度": score += 6; break;
            default: score += 2; break;
        }

        // 路线熟悉度权重5%
        score += Math.min(routeFamiliarity / 10.0, 1.0) * 5;

        return score;
    }

    /**
     * 创建司机记录
     */
    private String createDriverRecord(String pickupDatetime, String pickupLocationId, String dropoffLocationId,
                                      int pickupHour, String workShift, String workDayType, double tripDistance,
                                      double totalAmount, double tripDuration, double avgSpeed, String drivingStyle,
                                      String incomeLevel, String efficiencyLevel, double hourlyIncome,
                                      double incomePerMile, int routeFamiliarity, String professionalismLevel,
                                      String workIntensity, String anomalyLevel) {

        return String.format("%s|%s|%s|%d|%s|%s|%.2f|%.2f|%.2f|%.2f|%s|%s|%s|%.2f|%.2f|%d|%s|%s|%s",
                pickupDatetime, pickupLocationId, dropoffLocationId, pickupHour, workShift,
                workDayType, tripDistance, totalAmount, tripDuration, avgSpeed, drivingStyle,
                incomeLevel, efficiencyLevel, hourlyIncome, incomePerMile, routeFamiliarity,
                professionalismLevel, workIntensity, anomalyLevel);
    }

    /**
     * 创建异常记录
     */
    private String createAnomalyRecord(String driverId, String pickupDatetime, int anomalyScore,
                                       int speedAnomaly, int fareAnomaly, int durationAnomaly,
                                       double avgSpeed, double totalAmount, double tripDuration, String drivingStyle) {

        return String.format("%s|%s|%d|%d|%d|%d|%.2f|%.2f|%.2f|%s",
                driverId, pickupDatetime, anomalyScore, speedAnomaly, fareAnomaly,
                durationAnomaly, avgSpeed, totalAmount, tripDuration, drivingStyle);
    }

    /**
     * 创建排名记录
     */
    private String createRankRecord(String driverId, double hourlyIncome, double incomePerMile,
                                    String efficiencyLevel, String professionalismLevel, String workIntensity,
                                    double tripDistance, double totalAmount, int routeFamiliarity, double efficiencyScore) {

        return String.format("%s|%.2f|%.2f|%s|%s|%s|%.2f|%.2f|%d|%.2f",
                driverId, hourlyIncome, incomePerMile, efficiencyLevel, professionalismLevel,
                workIntensity, tripDistance, totalAmount, routeFamiliarity, efficiencyScore);
    }

    /**
     * 创建模式记录
     */
    private String createPatternRecord(String pickupLocationId, String dropoffLocationId, int pickupHour,
                                       String workShift, String drivingStyle, double tripDistance, double totalAmount) {

        return String.format("%s|%s|%d|%s|%s|%.2f|%.2f",
                pickupLocationId, dropoffLocationId, pickupHour, workShift,
                drivingStyle, tripDistance, totalAmount);
    }

    /**
     * 创建工作模式记录
     */
    private String createWorkPatternRecord(String driverId, int pickupHour, int pickupWeekday,
                                           String drivingStyle, String efficiencyLevel, String professionalismLevel) {

        return String.format("%s|%d|%d|%s|%s|%s",
                driverId, pickupHour, pickupWeekday, drivingStyle,
                efficiencyLevel, professionalismLevel);
    }

    /**
     * 创建司机Join记录
     */
    private String createDriverJoinRecord(String pickupLocationId, String dropoffLocationId, int routeFamiliarity,
                                          double tripDistance, double totalAmount, double hourlyIncome,
                                          String professionalismLevel) {

        return String.format("%s|%s|%d|%.2f|%.2f|%.2f|%s",
                pickupLocationId, dropoffLocationId, routeFamiliarity,
                tripDistance, totalAmount, hourlyIncome, professionalismLevel);
    }

    /**
     * 创建地点Join记录
     */
    private String createLocationJoinRecord(String driverId, String dropoffLocationId, int routeFamiliarity,
                                            double tripDistance, double totalAmount, String professionalismLevel) {

        return String.format("%s|%s|%d|%.2f|%.2f|%s",
                driverId, dropoffLocationId, routeFamiliarity,
                tripDistance, totalAmount, professionalismLevel);
    }

    private boolean isValidData(String driverId, String pickupLocationId, String dropoffLocationId) {
        return !driverId.isEmpty() && !pickupLocationId.isEmpty() && !dropoffLocationId.isEmpty() &&
                !driverId.equals("null") && !pickupLocationId.equals("null") && !dropoffLocationId.equals("null");
    }

    private String[] parseCSVLine(String line) {
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    private String[] parseQuotedCSV(String line) {
        List<String> fields = new ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }

    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }
}
